-module(rtps_reader).

%%% @doc RTPS Endpoint representing the objects that can be used to
%%% receive messages communicating CacheChanges.

-behaviour(gen_statem).

%% -include_lib("kernel/include/logger.hrl").
-include("rtps.hrl").
-include("rtps_history_cache.hrl").
-include("rtps_sedp.hrl").

%% API 

%% TODO: The functions get_/remove_changes should not be part of
%% the reader implementtaion. In the specs, DDS directly calls the
%% history cache for doing this.
-export([start_link/4,
	 matched_writer_add/2, matched_writer_add/3, matched_writer_remove/2,
	 available_changes/1, remove_change/2, get_qos/1,
	 info/1]).

%% gen_statem callbacks
-export([init/1, callback_mode/0,
	 stateless/3, 
	 stateful/3, 
	 handle_event/4, terminate/3, code_change/4]).

-record(data, {sup :: pid(), opts = [] :: [proplists:property()], sedp :: pid(),
	       guid :: rtps:guid(), reader_cache :: pid() | undefined,
	       u_locs :: [rtps:locator()] | undefined, m_locs :: [rtps:locator()],
	       reliability_level :: rtps:reliability_level(), topic_kind :: rtps:topic_kind(),
	       expects_inline_qos :: boolean(), qos :: rtps_qos:qos(),
	       reg :: pid() | undefined, proxies_sup :: pid() | undefined}).

%%%===================================================================
%%% API
%%%===================================================================

start_link(Sup, Entity_id, Reg, Opts) ->
    gen_statem:start_link({via, rtps_reg, {Reg, Entity_id}}, ?MODULE, [Sup, Opts], []).

-spec available_changes(Pid) -> [Change] when
      Pid :: pid(),
      Change :: rtps_history_cache:cache_change().
%% @doc Changes returned to the application remain in the RTPS
%% Reader’s HistoryCache such that subsequent read or take operations
%% can find them again. This is part of the read operation. See
%% 8.2.9.2.2 Transition T2 for details.
available_changes(Pid) ->
    {ok, Reader_cache} = gen_statem:call(Pid, reader_cache),
    rtps_history_cache:available_changes(Reader_cache).

-spec remove_change(Pid, Change) -> ok when
      Pid :: pid(),
      Change :: rtps_history_cache:cache_change().
%% @doc This operation indicates that the details regarding the CacheChange
%% need not be maintained in the HistoryCache.
remove_change(Pid, Change) ->
    {ok, Reader_cache} = gen_statem:call(Pid, reader_cache),
    rtps_history_cache:remove_change(Reader_cache, Change).

-spec matched_writer_add(Pid, Writer) -> Result when
      Pid :: pid(),
      Writer :: rtps:guid(),
      Result :: supervisor:startchild_ret() | {error, Reason},
      Reason :: term(). 
%% @doc This operation adds the WriterProxy a_writer_proxy to the
%% StatefulReader::matched_writers..
matched_writer_add(Pid, Writer) ->
    matched_writer_add(Pid, Writer, []).

-spec matched_writer_add(Pid, Writer, Opts) -> Result when
      Pid :: pid(),
      Writer :: rtps:guid(),
      Opts :: [proplists:property()],
      Result :: supervisor:startchild_ret() | {error, Reason},
      Reason :: term(). 
%% @doc This operation adds the WriterProxy a_writer_proxy to the
%% StatefulReader::matched_writers..
matched_writer_add(Pid, Writer, Opts) ->
    gen_statem:call(Pid, {matched_writer_add, Writer, Opts}).

-spec matched_writer_remove(Pid, Writer) -> ok when
      Pid :: pid(),
      Writer :: rtps:guid().
%% @doc This operation removes the WriterProxy a_writer_proxy from the
%% set StatefulReader::matched_writers.
matched_writer_remove(Pid, Writer) ->
    gen_statem:cast(Pid, {matched_writer_remove, Writer}).

-spec get_qos(Pid) -> Qos | undefined when
    Pid :: pid(),
    Qos :: [rtps:parameter()].
%% @doc This operation allows access to the existing set of QoS
%% policies for the reader.
get_qos(Pid) ->
    gen_statem:call(Pid, get_qos).

info(Pid) ->
    gen_statem:call(Pid, info).

%%%===================================================================
%%% gen_statem callbacks
%%%===================================================================

%% @private
init([Sup, Opts]) ->
    Reliability_level = proplists:get_value(reliability_level, Opts),
    case proplists:get_value(state_type, Opts) of
	%% According to 8.4.11.2, a stateless reader can not support
	%% the reliable protocol
	stateless when Reliability_level =:= reliable ->
	    {stop, badarg};
	State_type ->
	    Guid_prefix = proplists:get_value(guid_prefix, Opts),
	    Entity_id = proplists:get_value(entity_id, Opts),
	    Guid = #guid{guid_prefix = Guid_prefix, entity_id = Entity_id},
	    U_locs = proplists:get_value(uniicast_locator_list, Opts, []),
	    M_locs = proplists:get_value(multicast_locator_list, Opts),
	    Topic_kind = proplists:get_value(topic_kind, Opts),
	    Reader_cache = proplists:get_value(history_cache, Opts),
	    Expects_inline_qos = proplists:get_value(expects_inline_qos, Opts, false),
	    Qos = proplists:get_value(qos, Opts, []),
	    case Entity_id of
		#entity_id{type = Type} when Type =/= built_in ->
		    %% Only insert endpoints in the SEDP which are of type
		    %% user. It may be also correct to insert endpoints of
		    %% type vendor and unknown, but never built_in or one and
		    %% endpoint with one of the predefined entity ids, which
		    %% will most likely even result in a deadlock situation
		    %% here.
		    process_flag(trap_exit, true),
		    Sedp = proplists:get_value(sedp, Opts),
		    Reader_proxy = #reader_proxy{remote_reader_guid = Guid,
						 expects_inline_qos = Expects_inline_qos,
						 unicast_locator_list = U_locs,
						 multicast_locator_list = M_locs},
		    ok = rtps_sedp:insert(Sedp, Reader_proxy, Qos);
		_ ->
		    Sedp = undefined
	    end,
	    Data = #data{sup = Sup, opts = Opts, guid = Guid, sedp = Sedp,
			 u_locs = U_locs, m_locs = M_locs, reliability_level = Reliability_level,
			 topic_kind = Topic_kind, reader_cache = Reader_cache,
			 expects_inline_qos = Expects_inline_qos, qos = Qos},
	    {ok, State_type, Data}
    end.

%% @private
callback_mode() ->
    [state_functions, state_enter].

%% @private
stateless(enter, _Old_state, #data{sup = Sup, opts = Opts, guid = Guid,
				   u_locs = U_locs, m_locs = M_locs,
				   reader_cache = Reader_cache} = Data) ->
    Reader_cache1 = reader_cache(Sup, Reader_cache),
    Transports_sup = proplists:get_value(transports_sup, Opts),
    Locs = case U_locs of
	       undefined -> M_locs;
	       _ -> M_locs ++ U_locs
	   end,
    [rtps_transport_sup:add(Transports_sup, self(), Locator, Guid, unknown, Opts) || Locator <- Locs],

    %% TODO: remove the following line
    %% ok = rtps_history_cache:add_sup_handler(Reader_cache1, rtps_reader_evt, []),

    {keep_state, Data#data{reader_cache = Reader_cache1}};
%% 8.4.11.1.2 Transition T2 DATA message is received. A stateless
%% reader receives submessages from the transports it is associated
%% with directly. These submessages may come from one or more remote
%% writers, be duplicates, out of sequence etc. Just stuff them into
%% the local history cache which will deal with duplicates by ignoring
%% them.
stateless(info, {rcv, Rcvs},
	  #data{reader_cache = Reader_cache, expects_inline_qos = Expects_inline_qos, qos = Qos} = Data) ->
    Fun = fun(#msg{src_guid = Src_guid, timestamp = Timestamp, size = Size,
		   submsg = #data_msg{writer_sn = Seq_num, serialized_data = Serialized_data,
				      serialized_key = Serialized_key,
				      inline_qos = Inline_qos}}) ->
		  Cache_change = #cache_change{sequence_number = Seq_num, writer_guid = Src_guid,
					       data_value = Serialized_data,
					       instance_handle = Serialized_key,
					       timestamp = Timestamp,
					       size = Size},
		  {true, {Cache_change, Inline_qos}};
	     (_Rcv) ->
		  false
	  end,
    %% TODO: add cache_change kind and instance_handle
    {Cache_changes, Inline_qoss} = lists:unzip(lists:filtermap(Fun, Rcvs)),
    rtps_history_cache:add_change(Reader_cache, Cache_changes),
    Qos1 = case Expects_inline_qos of
	       true ->
		   merge_inline_qos([Qos | Inline_qoss]);
	       false ->
		   Qos
    end,
    {keep_state, Data#data{qos = Qos1}};
stateless({call, From}, get_qos, #data{qos = Qos}) ->
    Reply = Qos,
    {keep_state_and_data, [{reply, From, Reply}]};
stateless(Event_type, Event_content, Data) ->
    handle_event(Event_type, Event_content, stateless, Data).

%% @private
stateful(enter, _Old_state, #data{sup = Sup, reader_cache = Reader_cache} = Data) ->
    Reader_cache1 = reader_cache(Sup, Reader_cache),
    Children = supervisor:which_children(Sup),
    {ok, Reg} = child(Children, registry),
    {ok, Proxies_sup} = child(Children, proxies),

    %% TODO: remove the following line
    %% ok = rtps_history_cache:add_sup_handler(Reader_cache1, rtps_reader_evt, []),

    {keep_state, Data#data{reader_cache = Reader_cache1, reg = Reg, proxies_sup = Proxies_sup}};
stateful({call, From}, {matched_writer_add, Writer, Extra_opts},
	 #data{reg = Reg, proxies_sup = Proxies_sup, opts = Opts, guid = Guid, reader_cache = Reader_cache}) ->
    Spec = #{id => Writer,
	     start => {rtps_writer_proxy, start_link, [Writer, Reg, Guid, Reader_cache, Extra_opts ++ Opts]},
	     restart => permanent,
	     type => worker},
    Reply = rtps_ofo_sup:start_child(Proxies_sup, Spec),
    %% TODO: update the SEDP (WHERE TO GET THE LOCS!?!?!
    {keep_state_and_data, [{reply, From, Reply}]};
stateful(cast, {matched_writer_remove, Writer}, #data{proxies_sup = Proxies_sup}) ->
    rtps_ofo_sup:stop_child(Proxies_sup, Writer),
    %% TODO: update the SEDP
    keep_state_and_data;
stateful({call, From}, get_qos, #data{reg = Reg}) ->
    Proxies = rtps_reg:which(Reg),
    Inline_qoss = [rtps_writer_proxy:get_qos(Pid) || {_, Pid} <- Proxies],
    Reply = merge_inline_qos(Inline_qoss),
    {keep_state_and_data, [{reply, From, Reply}]};
stateful(Event_type, Event_content, Data) ->
    handle_event(Event_type, Event_content, stateful, Data).

%% @private
handle_event({call, From}, reader_cache, _State_name, #data{reader_cache = Reader_cache}) ->
    Reply = {ok, Reader_cache},
    {keep_state_and_data, [{reply, From, Reply}]};
handle_event({call, From}, info, State_name,
	     #data{reg = Reg, guid = #guid{entity_id = Entity_id},
		   reliability_level = Reliability_level, topic_kind = Topic_kind,
		   u_locs = U_locs, m_locs = M_locs}) ->
    L = case State_name of
	    stateless ->
		[{locators, [Locator || {Locator, _Pid} <- rtps_reg:which(Reg)]},
		 {u_locs, U_locs}, {m_locs, M_locs}];
	    stateful ->
		[{proxies, [Proxy || {Proxy, _Pid} <- rtps_reg:which(Reg)]}]
	       end,
    Reply = [{entity_id, Entity_id}, {reliability_level, Reliability_level},
	     {topic_kind, Topic_kind}] ++ L,
    {keep_state_and_data, [{reply, From, Reply}]};

%% TODO: Sink unhadled events. Remove these clauses.
handle_event({call, From} = _Event_type, _Event_content, _State, _Data) ->
    %% ?LOG_WARNING("Unhandled event: Type ~p, Content ~p", [Event_type,  Event_content]),
    Reply = ok,
    {keep_state_and_data, [{reply, From, Reply}]};
handle_event(_Event_type,  _Event_content, _State, _Data) ->
    %% ?LOG_WARNING("Unhandled event: Type ~p, Content ~p", [Event_type,  Event_content]),
    keep_state_and_data.

%% @private
terminate(_Reason, _State, #data{sedp = Sedp, guid = Guid}) ->
    rtps_sedp:delete(Sedp, #reader_proxy{remote_reader_guid = Guid}),
    ok.

%% @private
code_change(_Old_vsn, State, Data, _Extra) ->
    {ok, State, Data}.

%%%===================================================================
%%% Internal functions
%%%===================================================================

%% @private
child(Children, Id) ->
    case lists:keyfind(Id, 1, Children) of
	{_Id, Child, _Type, _Modules} when is_pid(Child) ->
	    {ok, Child};
	{_Id, Child, _Type, _Modules} when Child =:= restarting ->
	    {error, Child};
	_ ->
	    {error, notfound}
    end.

%% @private
%% @doc Add a history cache to the endpoint's supervisor if
%% Reader_cache is not undefined.
reader_cache(Sup, undefined) ->
    {ok, Pid} = rtps_endpoint_sup:history_cache(Sup),
    Pid;
reader_cache(_Sup, Reader_cache) ->
    Reader_cache.

merge_inline_qos(Inline_qoss) ->
    merge_inline_qos(Inline_qoss, []).

merge_inline_qos([], Acc) ->
    Acc;
merge_inline_qos([L | Rest], Acc) ->
    L1 = lists:keysort(#parameter.id, L),
    Acc1 = lists:keymerge(#parameter.id, Acc, L1),
    merge_inline_qos(Rest, Acc1).
